package com.zte.data.migrate;

import com.google.common.io.Files;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Base64;

public class MigrateApplication {

    public static final String SERVER_IP = "localhost";
    public static final int SERVER_PORT = 9022;

    public static class MySimpleTcpClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf buf) throws Exception {
            System.out.println("receive buf ===> " + ByteBufUtil.hexDump(buf));
        }
    }


    public static void main(String[] args) {
        System.out.println("hello world");


        Bootstrap bootstrap = new Bootstrap();
        EventLoopGroup group = new NioEventLoopGroup();
        bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true);

        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast("receiveHandler", new MySimpleTcpClientHandler());
            }
        });

        try {
            ChannelFuture future = bootstrap.connect(SERVER_IP, SERVER_PORT).sync();

            Files.fileTreeTraverser().preOrderTraversal(new File("/Users/linxy/Documents/zte/Data/2017")).forEach(f -> {
                System.out.println(f.getAbsolutePath());
                if (f.isFile()) {
                    try {
                        Files.readLines(f, Charset.defaultCharset()).forEach(line -> {
                            String[] values = line.split(",");
                            if (values.length == 2) {
                                ByteBuf buf = Unpooled.wrappedBuffer(Base64.getDecoder().decode(values[1]));
                                System.out.println(ByteBufUtil.hexDump(buf));
                                future.channel().writeAndFlush(buf, future.channel().voidPromise());
                                try {
                                    // 不要发送太快
                                    Thread.sleep(2);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            } else {
                                System.out.println("read line format error: " + line);
                            }
                        });
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });

            future.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }

    }
}
